<?php
// +----------------------------------------------------------------------
// | 消息队列
// +----------------------------------------------------------------------

namespace helper;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

class RabbitMQ
{
    private array $config;

    /**
     * 队列名
     * @var string $queue
     */
    private string $queue;

    private $channel;

    private $connection;




    // 1) composer require php-amqplib/php-amqplib

    // 2) 编写代码
    // receive.php
    /**
     *
        require_once __DIR__ . '/../vendor/autoload.php';

        use helper\RabbitMQ;

        $mq = new RabbitMQ([], 'hello');

        echo " [*] Waiting for messages. To exit press CTRL+C\n";
        $callback = function ($msg) {
            echo ' [x] Received ', $msg->body, "\n";
        };
        $mq->receive($callback);
     *
     */

    // send.php
    /**
        require_once __DIR__ . '/../vendor/autoload.php';

        use helper\RabbitMQ;

        $mq = new RabbitMQ([], 'hello');

        $mq->send('Hello World!', 'hello');
        echo " [x] Sent 'Hello World!'\n";
     */


    // 3) 终端运行脚本
    // php receive.php
    // php send.php

    /**
     * 创建连接和channel
     * @param array $config 配置
     * @param string $queue 队列名
     * @return void
     */
    public function __construct(array $config = [], string $queue = '')
    {
        // 连接服务器
        $this->connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
        // 连接通道
        $this->channel = $this->connection->channel();
        // 声明队列
        $this->queue = $queue;
        $this->channel->queue_declare($queue, false, false, false, false);
    }

    /**
     * 发送消息
     * @param string $body
     * @param string $routing_key 路由名
     */
    public function send(string $body, string $routing_key)
    {
        $msg = new AMQPMessage($body);
        $this->channel->basic_publish($msg, '', $routing_key);
    }

    /**
     * 接收消息
     */
    public function receive($callback)
    {
        $this->channel->basic_consume($this->queue, '', false, true, false, false, $callback);
        while ($this->channel->is_open()) {
            $this->channel->wait();
        }
    }

    /**
     * 销毁
     * @return void
     * @throws \Exception
     */
    public function __destroy()
    {
        //关闭通道和连接
        $this->channel->close();
        $this->connection->close();
    }
}